from flask import Flask, send_from_directory
from flask_socketio import SocketIO
from kafka import KafkaConsumer
import threading
import webbrowser
import time
import os

# 创建 Flask 应用
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your_secret_key'
socketio = SocketIO(app, cors_allowed_origins="*")  # 启用跨域

# Kafka 配置
KAFKA_BROKER = '192.168.235.128:9092'  # Kafka Broker 地址
KAFKA_TOPIC = 'test3'  # 替换为你的 Kafka Topic 名称

# 指定 HTML 文件路径
BASE_DIR = os.path.dirname(os.path.abspath(__file__))  # 当前脚本所在目录
STATIC_DIR = os.path.join(BASE_DIR, "static")  # 静态文件夹
INDEX_HTML = "index1.html"  # HTML 文件名

# 消费 Kafka 数据并推送给 WebSocket 客户端
def kafka_consumer():
    """Kafka 消费者线程"""
    consumer = KafkaConsumer(
        KAFKA_TOPIC,
        bootstrap_servers=[KAFKA_BROKER],
        group_id='socket-io-group',  # 消费者组
        auto_offset_reset='earliest'  # 从最早的消息开始读取
    )
    print("Kafka Consumer started and listening for messages...")
    for message in consumer:
        # 获取 Kafka 消息内容
        data = message.value.decode('utf-8')
        print(f"Received from Kafka: {data}")

        # 将消息通过 WebSocket 广播给所有客户端
        socketio.emit('kafka-message', {'data': data})
        print(f"Broadcasted to WebSocket: {data}")

# WebSocket 事件
@socketio.on('connect')
def handle_connect():
    print("Client connected")

@socketio.on('disconnect')
def handle_disconnect():
    print("Client disconnected")

# 提供 HTML 页面
@app.route("/")
def index():
    """提供 index.html 页面"""
    return send_from_directory(STATIC_DIR, INDEX_HTML)

# 自动打开浏览器
def open_browser():
    """自动打开浏览器访问 Flask 服务"""
    time.sleep(1)  # 确保 Flask 服务已启动
    webbrowser.open("http://localhost:5000")  # 访问本地的 Flask 服务

if __name__ == '__main__':
    # 在后台线程中启动 Kafka 消费者
    kafka_thread = threading.Thread(target=kafka_consumer)
    kafka_thread.daemon = True  # 设置为守护线程，主线程结束时自动退出
    kafka_thread.start()

    # 启动 Flask-SocketIO 服务线程
    flask_thread = threading.Thread(target=lambda: socketio.run(app, host='0.0.0.0', port=5000, allow_unsafe_werkzeug=True))
    flask_thread.daemon = True
    flask_thread.start()

    # 自动打开浏览器
    open_browser()

    # 保持主线程运行
    flask_thread.join()
